package com.cssrc.wenjie.hbase.service.intf;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.cssrc.wenjie.hbase.service.HBaseDML;
import com.cssrc.wenjie.hbase.util.HBaseUtil;

/**   
* @des Hbase行数据的实现类
* @author wenjie   
* @date 2018年8月31日 下午5:55:39 
* @version V1.0  
*/
public abstract class HBaseDMLImpl implements HBaseDML {
    public static Logger log = Logger.getLogger(HBaseDMLImpl.class);
    
    private static Connection conn = null;
    static {
        Configuration conf = HBaseConfiguration.create();
        try {
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    public abstract Class<?> getClassName();
    
    @Override
    public <T> List<T> get(String tableName, String... rowkeys) {
        List<T> list = new ArrayList<T>();
        try {
            HTable table = (HTable)conn.getTable(TableName.valueOf(tableName));
            List<Get> gets = new ArrayList<Get>();
            for (String rowkey : rowkeys) {
                if (StringUtils.isEmpty(rowkey)) {
                    continue;
                }
                Get get = new Get(Bytes.toBytes(rowkey));
                gets.add(get);
            }
            Result[] results = table.get(gets);
            for (int i = 0; i < results.length; i++) {
                if (results[i] == null) {
                    continue;
                }
                T bean = (T)getClassName().newInstance();
                Result result = results[i];
                if (result == null || result.isEmpty()) {
                    continue;
                }
                bean = HBaseUtil.result2Bean(result, bean);
                list.add(bean);
            }
        } catch (Exception e) {
            log.error("执行查询失败;", e);
        }
        return list;
    }
    
    @Override
    public void delete(String tableName, String... rowkeys) {
        try {
            HTable table = (HTable)conn.getTable(TableName.valueOf(tableName));
            List<Delete> deletes = new ArrayList<Delete>();
            for (String rowkey : rowkeys) {
                if (StringUtils.isEmpty(rowkey)) {
                    continue;
                }
                deletes.add(new Delete(Bytes.toBytes(rowkey)));
            }
            table.delete(deletes);
        } catch (Exception e) {
            log.error("执行删除失败;", e);
        }
    }
    
    @Override
    public <T> void insert(T... objs) {
        Assert.notEmpty(objs, "带插入数据不能为null");
        List<Put> puts = new ArrayList<Put>();
        try {
            String tableName = HBaseUtil.getTableName(objs[0]);
            HTable table = (HTable)conn.getTable(TableName.valueOf(tableName));
            for (Object obj : objs) {
                if (obj == null) {
                    continue;
                }
                Put put = HBaseUtil.bean2Put(obj);
                puts.add(put);
            }
            table.put(puts);
        } catch (Exception e) {
            log.error("数据插入失败", e);
        }
    }
}
